package com.mininglamp.com.mininglamp.recommender

import com.mongodb.casbah.commons.MongoDBObject
import com.mongodb.casbah.{MongoClient, MongoClientURI}
import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
  * Project: ECommerceRecommendSystem
  * Package: com.mininglamp.com.mininglamp.recommender
  * Description:数据加载服务（data -> mongodb）
  *
  * Created by ZhouPeng on 2021/12/23 11:40
  **/
object DataLoader {

  //定义数据路径
  val PRODUCTS_DATA_PATH = "D:\\ProjectCode\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\products.csv"
  val RATINGS_DATA_PATH = "D:\\ProjectCode\\ECommerceRecommendSystem\\recommender\\DataLoader\\src\\main\\resources\\ratings.csv"
  //定义mongo中的表名
  val MONGODB_PRODUCT_COLLECTION = "Product"
  val MONGODB_RATING_COLLECTION = "Rating"

  def main(args: Array[String]): Unit = {

    //基本配置
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://master:27017/recommender",
      "mongo.db" -> "recommender"
    )

    val sparkConf = new SparkConf().setMaster(config("spark.cores")).setAppName("DataLoader")

    val sparkSession = SparkSession.builder().config(sparkConf).getOrCreate()

    //todo 处理数据
    val productRDD = sparkSession.sparkContext.textFile(PRODUCTS_DATA_PATH)
    val ratingRDD = sparkSession.sparkContext.textFile(RATINGS_DATA_PATH)


    import sparkSession.implicits._

    val productDF = productRDD.map(
      data => {
        val dataArrays = data.split("\\^")
        Product(dataArrays(0).toInt, dataArrays(1).toString, dataArrays(4).toString,
          dataArrays(5).toString, dataArrays(6).toString)
      }
    ).toDF()

    val ratingDF = ratingRDD.map(
      data => {
        val dataArrays = data.split(",")
        Rating(dataArrays(0).toInt, dataArrays(1).toInt, dataArrays(2).toDouble, dataArrays(3).toLong)
      }
    ).toDF()


    /**
      * todo 设置隐式mongodb配置变量。
      * todo 好处就是在调用写入mongo方法获取mongo客户端时，就不需要每次将mongo的配置作为参数传入了。
      * todo 方法将该参数设置为隐式参数，调用时就可以省略，编译器会按照隐式参数类型在全局作用域内寻找对应的隐式值，与隐式值的名称无关。有点类似于默认值。
      * todo 但是隐式参数优先级高于默认值。并且同一个作用域内同类型的隐式变量只能有1个
      */
    implicit val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    //todo 写数据到mongodb
    storeData2Mongdb(productDF, ratingDF)
  }


  /**
    *
    * @param productDF
    * @param ratingDF
    * @param mongoConfig mongo配置设置为隐式参数。作用：编译器会在省略隐式参数的情况下去全局作用域内找到隐式值作为缺省参数
    */
  def storeData2Mongdb(productDF: DataFrame, ratingDF: DataFrame)(implicit mongoConfig: MongoConfig): Unit = {

    //获取mongodb连接客户端
    val mongoClient = MongoClient(MongoClientURI(mongoConfig.mongoUri))
    //定义要操作的表(collection)
    val productCollection = mongoClient(mongoConfig.mongodb)(MONGODB_PRODUCT_COLLECTION)
    val ratingCollection = mongoClient(mongoConfig.mongodb)(MONGODB_RATING_COLLECTION)

    //如果表存在，则删除
    productCollection.dropCollection()
    ratingCollection.dropCollection()

    //todo 采用spark原生方式写入mongodb
    productDF.write
      .option("uri", mongoConfig.mongoUri)
      .option("collection", MONGODB_PRODUCT_COLLECTION)
      //以mongo格式写入
      .format("com.mongodb.spark.sql")
      .mode("overwrite")
      .save()

    ratingDF.write
      .option("uri", mongoConfig.mongoUri)
      .option("collection", MONGODB_RATING_COLLECTION)
      .mode("overwrite")
      .format("com.mongodb.spark.sql")
      .save()

    //todo 给表创建索引,加速mongo查询
    productCollection.createIndex(MongoDBObject("productId" -> 1))
    ratingCollection.createIndex(MongoDBObject("userId" -> 1))
    ratingCollection.createIndex(MongoDBObject("productId" -> 1))

    //关闭mongo客户端
    mongoClient.close()
  }
}


/**
  * product样例类
  *
  * 6797
  * PHILIPS飞利浦
  * HQ912/15两刀头充电式电动剃须刀
  * 222,621,691B002TKLK0S
  * https://images-cn-4.ssl-images-amazon.com/images/I/415UjOLnBML._SY300_QL70_.jpg
  * 家用电器|个人护理电器|电动剃须刀
  * 飞利浦|剃须刀|家用电器|好用|外观漂亮
  */
case class Product(productId: Int, name: String, imageUrl: String, categories: String, tags: String)

/**
  * rating样例类
  * 4867
  * 457976
  * 5.0
  * 1395676800
  */
case class Rating(userId: Int, productId: Int, score: Double, timestamp: Long)

/**
  * mongodb配置样例类
  *
  * @param mongoUri mongo连接uri
  * @param mongodb  mongo连接数据库
  */
case class MongoConfig(mongoUri: String, mongodb: String)
